package cn.hfuu.flink;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.CloseableIterator;

import java.time.Duration;

/**
 * @author oldliu
 * @since 1.0
 */
public class WatserSensorWindow3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(4);
        //connector
        DataStreamSource<String> source = env.socketTextStream("hadoopmaster", 7777);
        WindowedStream<WaterSensor, String, TimeWindow> window = source.map(line -> {
            String[] split = line.split(",");
            if (split.length == 3) {
                return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
            }
            return null;
        }).filter(x -> x != null).keyBy(x -> x.getId()).window(TumblingProcessingTimeWindows.of(Duration.ofMinutes(2)));
        SingleOutputStreamOperator<WaterSensor> r = window.reduce((x, y) -> {
            return new WaterSensor(x.getId(), Math.max(x.getTs(), y.getTs()), x.getVc() + y.getVc());
        });

        ////
        // 聚合状态类，用于存储总和和计数
        class AverageAccumulator {
            public double sum;
            public long count;

            public AverageAccumulator() {
                this.sum = 0;
                this.count = 0;
            }

            public void add(int value) {
                sum += value;
                count++;
            }

            public double getAverage() {
                return count > 0 ? sum / count : 0;
            }
        }

////
//        SingleOutputStreamOperator<WaterSensor> max = window.maxBy("id");
//
//        SingleOutputStreamOperator<Double> agg = window.aggregate(new AggregateFunction<WaterSensor,
//                AverageAccumulator, Double>() {
//
//            @Override
//            public AverageAccumulator createAccumulator() {
//                return new AverageAccumulator();
//            }
//
//            @Override
//            public AverageAccumulator add(WaterSensor v, AverageAccumulator accumulator) {
//                accumulator.add(v.getVc());
//                return accumulator;
//            }
//
//            @Override
//            public Double getResult(AverageAccumulator o) {
//                return o.getAverage();
//            }
//
//            @Override
//            public AverageAccumulator merge(AverageAccumulator a, AverageAccumulator b) {
//                a.sum += b.sum;
//                a.count += b.count;
//                return a;
//            }
//        });
        System.out.println("累计水位：");
        //CloseableIterator<WaterSensor> list = r.collectAsync();

        r.print();

//        System.out.println("最大水位：");
//        max.print();
//        System.out.println("平均水位");
//        agg.print();
        env.execute();

    }
}
